package n81_自定义线程池;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import static utils.Sleeper.sleep;

/**
 * @author fu
 * @date 2022-02-22 21:48
 * 概要：
 * --------------------------------------------------------------------------------
 * |                                                                              |
 * |       线程池：创建一批线程 得到重复的利用  减少内存占用 和 线程数量 ；避免上下文切换   |
 * |                                                                             |
 * |==============================================================================
 * 线程不能建太多
 * 多了 超过CPU核心，会造成 某些线程获取不到CPU时间片而阻塞； 造成上下文切换 而性能降低
 *
 * 应该充分利用 已有线程的潜力
 *
 */
@Slf4j(topic = "c.Test")
public class T2_拒绝策略 {

    public static void main(String[] args) throws InterruptedException {

        final ThreadPool threadPool = new ThreadPool(2,1000, TimeUnit.MILLISECONDS, 1,
                // 最后一个 参数 是 策略
                (queue, task)->{
//                1) 死等
//                    queue.put(task);
//                2) 超时等待
//                    queue.offer(task, 500, TimeUnit.MILLISECONDS);  // 500 毫秒超时 小于 执行1s 最后一个不执行
                    queue.offer(task, 1500, TimeUnit.MILLISECONDS); // 1500 毫秒超时 大于 执行1s 正常执行
//                3) 让调用者 放弃任务执行
//                    log.debug("放弃执行{}", task);
//                4) 让调用者 抛出异常  【剩下的 任务不再执行】
//                    throw new RuntimeException("任务执行失败。。。"+task);
//                5) 让调用者 自己执行任务
//                    task.run();

        });

//=======================================================================================================
        for (int i = 0; i < 4; i++) {   // 15个任务， 任务队列10+线程池可执行2  【第十三个线程 [ put 会阻塞]】

            int j = i;
            threadPool.execute(()->{
                sleep(1);  // 执行时间拉长 只看 任务队列满的情况
                log.debug("{}", j);

            });
        }

    }
}

// 拒绝策略
@FunctionalInterface
interface RejectPolicy<T> {
    void reject(BlockingQueue<T> queue, T task);
}


// 线程池 类
@Slf4j(topic = "c.Pool")
class ThreadPool {

    // 任务队列
    private BlockingQueue<Runnable> taskQueue; // 【用来放入 任务】

    // 线程集合
    private HashSet<Worker> workers = new HashSet(); // 这里有共享数据 在 execute()方法里用锁

    // 核心线程数
    private int coreSize;

    // 设置超时时间，过一段时间 还没任务 则让线程结束
    private long pollTimeout;
    private TimeUnit timeUnit;

    // 拒绝策略
    RejectPolicy<Runnable> rejectPolicy;

    public ThreadPool(int coreSize, long pollTimeout, TimeUnit timeUnit, int queueCapcity, RejectPolicy<Runnable> rejectPolicy) {
        this.coreSize = coreSize;
        this.pollTimeout = pollTimeout;
        this.timeUnit = timeUnit;
        this.taskQueue = new BlockingQueue<>(queueCapcity);
        this.rejectPolicy = rejectPolicy;
    }

    // 执行任务 ： 来一个任务，如果 [线程集合]里任务数 <  核心数 ; 放入 [线程集合] 并且执行
    //            否则 放入 任务队列
    public void execute(Runnable task){

        synchronized (workers){

            // 1. 任务数 <  核心数 coreSize ， 交给 worker 创建线程对象 & 来执行任务
            if(workers.size() < coreSize){
                final Worker worker = new Worker(task);
                log.debug("新增 worker {}, task {}", worker, task);
                workers.add(worker); // [------ 执行完了 会移除 ------]

                worker.start();

                // 2. 如果超过 coreSize , 则加入 任务队列 暂存
            }else{
//                taskQueue.put(task);// 这里满了会阻塞 ； 执行完了会唤醒 take阻塞的
                /*
                1) 死等
                2) 超时等待
                3) 让调用者 放弃任务执行
                4) 让调用者 抛出异常
                5) 让调用者 自己执行任务
                 */
                taskQueue.tryPut(rejectPolicy, task);



            }

        }

    }

    // 线程包装
    class Worker extends Thread{

        private Runnable task;

        public Worker(Runnable task) {
            this.task = task;
        }

        /**
         * 要么从 初始化 里获得 任务执行
         * 要么从 任务队列里 获取 任务执行
         */
        @Override
        public void run() {

//HEAD>>>>>>>>>>>>>>>>>>>>>>>>>>>
//            while(task != null    ||    (task = taskQueue.take()) != null){// take() 会 【阻塞等待  没有任务时 线程不会停止】
//===============================
            while(task != null    ||    (task = taskQueue.poll(pollTimeout,timeUnit)) != null){// take()  【没有任务时 超时停止】
//END<<<<<<<<<<<<<<<<<<<<<<<<<<<<
                try{
                    log.debug("正在执行...{}", task);
                    task.run();

                } catch (Exception e){

                } finally {
                    task = null;
                }
            }
            // 移除 任务
            synchronized (workers) {
                log.debug("worker 被移除 {}",this);
                workers.remove(this);
            }

        }
    }


}

// 任务队列 类
@Slf4j(topic = "c.Block")
class BlockingQueue<T> {

    // 1. 任务队列  使用双向链表
    private Deque<T> queue = new ArrayDeque<>(); // ArrayDeque 通常 比 LinkedList 性能更好

    // 2. 锁     将来线程池都要从 任务队列的 头部 获取任务 【取任务时 一个一个获取，前面线程获取 后面得等待】
    private ReentrantLock lock = new ReentrantLock();

    // 3. 针对生产者的 条件变量 【任务时达到容量等待；不可以无限量的 往消息队列里 添加任务】
    private Condition fullWaitSet = lock.newCondition();

    // 4. 针对消费者的 条件变量 【没有任务时(队列空)等待】
    private Condition emptyWaitSet = lock.newCondition();

    // 5. 容量
    private int capcity;

    public BlockingQueue(int capcity) {
        this.capcity = capcity;
    }

    // 阻塞获取 --- 带超时
    public T poll(long timeOut, TimeUnit unit){
        lock.lock();

        try{
            // 1. 转成 纳秒
            long nanos = unit.toNanos(timeOut);
            // 没有 元素 --- 阻塞等待
            while(queue.isEmpty()){
                try {
                    // 小于等于0 就是等待超时了
                    if(nanos <= 0){
                        log.debug("...[头部获取]等待超时了...等不起了...");
                        return null;
                    }
                    nanos = emptyWaitSet.awaitNanos(nanos);// 返回 等待剩余时间，防止虚假唤醒 重复等待，而要让其等待剩余时间
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            // 有 元素 获取头部元素 返回
            final T t = queue.removeFirst();

            fullWaitSet.signal();// 【唤醒 , 可以往消息队列里 放入信息了】
            return t;

        } finally {

            lock.unlock();
        }

    }

    // 阻塞获取 --- 从队列中 获取元素
    public T take(){
        lock.lock();

        try{
            // 没有 元素 --- 阻塞等待
            while(queue.isEmpty()){
                try {
                    emptyWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            // 有 元素 获取头部元素 返回
            final T t = queue.removeFirst();

            fullWaitSet.signal();// 【唤醒 , 可以往消息队列里 放入信息了】
            return t;

        } finally {

            lock.unlock();
        }

    }
//=====================================================================================================================
    // 阻塞添加 --- 从队列尾部 添加元素   【--- 死等 ---】
    public void put(T task){
        lock.lock();

        try{
            // 队列 容量已满 --- 阻塞等待
            while(size()>=capcity){
                try {
                    log.debug("put  等待加入队列 task {}...", task);
                    fullWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            log.debug("加入队列 task {}", task);
            // 未满 则存入 ,并且 唤醒上面阻塞的
            queue.addLast(task);

            emptyWaitSet.signal();// 【唤醒 , 可以往消息队列里 拿出信息了】

        }finally {

            lock.unlock();
        }
    }
    // 阻塞添加 --- 从队列尾部 添加元素 【--- 带超时时间 ---】
    public boolean offer(T task, long offerTimeOut, TimeUnit unit){
        lock.lock();

        try{
            // 1. 转成 纳秒
            long nanos = unit.toNanos(offerTimeOut);

            // 队列 容量已满 --- 阻塞等待
            while(size()>=capcity){
                try {
                    if(nanos <= 0){
                        log.debug("...[尾部添加]等待超时了...等不起了...");
                        return false;
                    }

                    log.debug("offer 等待加入队列 task {}...", task);
                    nanos = fullWaitSet.awaitNanos(nanos);

                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            log.debug("加入队列 task {}", task);
            // 未满 则存入 ,并且 唤醒上面阻塞的
            queue.addLast(task);

            emptyWaitSet.signal();// 【唤醒 , 可以往消息队列里 拿出信息了】

            return true;
        }finally {

            lock.unlock();
        }
    }

    // 获取队列 大小 --- 获取队列里排队任务还有几个
    public int size(){
        lock.lock();

        try{

            return queue.size();

        }finally {
            lock.unlock();
        }

    }

    // 尝试 -- 任务 放入 队列 ； 队列满了 则使用 【--- 拒绝策略 ---】
    public void tryPut(RejectPolicy<T> rejectPolicy, T task) {
        lock.lock();

        try{
            if(queue.size() >= capcity){

                rejectPolicy.reject(this, task);

            }else{

                log.debug("加入队列 task {}", task);
                // 未满 则存入 ,并且 唤醒上面阻塞的
                queue.addLast(task);

                emptyWaitSet.signal();// 【唤醒 , 可以往消息队列里 拿出信息了】

            }


        }finally {
            lock.unlock();
        }

    }
}



